/*
 * @Author: hzm 
 * @Date: 2020-06-22 17:10:29 
 * @Last Modified by: hzm
 * @Last Modified time: 2020-06-22 20:13:16
 */

#include "LoopThreadPool.h"
#include <cassert>
#include "MLogManager.h"

using namespace HeyIM;

LoopThreadPool::LoopThreadPool(int threadNum): m_threadNum(threadNum) {
    m_status.store(Status::None);

    assert(threadNum > 0);

    for(int i = 0; i < threadNum; ++i) {
        EventLoopPtr p = std::make_shared<EventLoop>();
        m_loops.emplace_back(p);
    }

}

LoopThreadPool::~LoopThreadPool() {

    assert(m_status.load() == Status::Stopped);

    MLogInfo("LoopThreadPool::~LoopThreadPool");
}

int LoopThreadPool::start() {
    assert(m_status.load() == Status::None);

    m_status.store(Status::Running);

    for(auto loop: m_loops) {
        
        auto preCb = [this]() {
            ++m_threadRunNum;
        };
        loop->post(preCb);
        std::thread t([this, loop](){
            loop->run();
            ++m_threadStopNum;
        });
        t.detach();
        m_threads.emplace_back(std::move(t));
    }

    MLogInfo("LoopThreadPool::start");

    return 0;
}

void LoopThreadPool::stop() {

    if(m_status.load() == Status::Stopped) {
        MLogInfo("LoopThreadPool has been stoped");
        return;
    }

    m_status.store(Status::Stopped);
    for(auto loop: m_loops) {
        loop->post([loop]() mutable{
            loop->stop();
            loop.reset();
        });
    }

    while(m_threadStopNum.load() != (int)m_loops.size()) {
        std::this_thread::yield();
    }

    m_loops.clear();
    m_threads.clear();

    MLogInfo("LoopThreadPool::stop");
}

EventLoopPtr LoopThreadPool::nextLoop() {
    if(m_status.load() == Status::Stopped) {
        return nullptr;
    }

    return m_loops[m_nextLoopID++ % m_threadNum];
}

EventLoopPtr LoopThreadPool::loop(event_base* base) {
    for(auto loop: m_loops) {
        if(loop-> eventBase() == base) {
            return loop;
        }
    }

    return nullptr;
}